package drds.data_migrate.task_group.task.data_exchange.exchanger;


import drds.common.Author;
import drds.data_migrate.Authors;
import drds.data_migrate.task_group.task.data_exchange.record.Record;
import drds.data_migrate.task_group.task.data_exchange.record.TerminateRecord;
import drds.data_migrate.vo.Messages;
import drds.data_migrate.vo.State;
import drds.data_migrate.vo.TaskConfig;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.Validate;

import java.util.Collection;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;


/**
 * 统计和限速都在这里
 */
@Author(name = Authors.LI_YANG)
@Slf4j
public class RecordQueue
{


    @Setter
    @Getter
    protected int taskGroupId;
    @Setter
    @Getter
    protected int capacity;
    @Setter
    @Getter
    protected int byteCapacity;
    @Setter
    @Getter
    protected long byteSpeed; // bps: bytes/s
    @Setter
    @Getter
    protected long recordSpeed; // tps: records/s
    @Setter
    @Getter
    protected long flowControlInterval;
    @Setter
    @Getter
    protected volatile boolean isClosed = false;
    @Setter
    @Getter
    protected TaskConfig taskConfig = null;
    @Setter
    @Getter
    protected volatile long waitReaderTime = 0;
    @Setter
    @Getter
    protected volatile long waitWriterTime = 0;
    @Setter
    @Getter
    private State state;
    @Setter
    @Getter
    private State lastState = new State();
    @Setter
    @Getter
    private int bufferSize = 0;
    @Setter
    @Getter
    private ArrayBlockingQueue<Record> recordQueue = null;
    @Setter
    @Getter
    private ReentrantLock lock;
    @Setter
    @Getter
    private Condition notInsufficient, notEmpty;

    public RecordQueue(final TaskConfig taskConfig)
    {
        //channel的queue里默认record为1万条。原来为512条
        int capacity = taskConfig.getTransportChannelCapacityByte();//2048
        long byteSpeed = taskConfig.getTransportChannelSpeedByte();//1024 * 1024
        long recordSpeed = taskConfig.getTransportChannelSpeedRecord();//10000

        if (capacity <= 0)
        {
            throw new IllegalArgumentException(String.format(
                    "通道容量[%d]必须大于0.", capacity));
        }


        this.taskGroupId = taskConfig.getTaskGroupId();
        this.capacity = capacity;
        this.byteSpeed = byteSpeed;
        this.recordSpeed = recordSpeed;
        this.flowControlInterval = taskConfig.getTransportChannelFlowcontrolinterval();//1000
        //channel的queue默认大小为8M，原来为64M
        this.byteCapacity = taskConfig.getTransportChannelCapacityByte();//8 * 1024 * 1024
        this.taskConfig = taskConfig;

        this.recordQueue = new ArrayBlockingQueue<Record>(this.getCapacity());
        this.bufferSize = taskConfig.getRecordListSize();

        lock = new ReentrantLock();
        notInsufficient = lock.newCondition();
        notEmpty = lock.newCondition();
    }


    public void setState(final State state)
    {
        this.state = state;
        this.lastState.reset();
    }

    public void push(final Record record)
    {
        Validate.notNull(record, "record不能为空.");
        this.put(record);
        this.statPush(1L);
    }

    public void pushTerminateRecord(final TerminateRecord terminateRecord)
    {
        Validate.notNull(terminateRecord, "record不能为空.");
        this.put(terminateRecord);


    }

    public void pushAll(final Collection<Record> recordCollection)
    {
        Validate.notNull(recordCollection);
        Validate.noNullElements(recordCollection);
        //
        this.addAll(recordCollection);
        this.statPush(recordCollection.size());
    }

    public Record pull()
    {
        Record record = this.take();
        this.statPull(1L);
        return record;
    }

    public void pullAll(final Collection<Record> recordCollection)
    {
        Validate.notNull(recordCollection);
        this.drainTo(recordCollection);
        this.statPull(recordCollection.size());
    }

    private void statPush(long recordSize)
    {
        state.increase(Messages.read_succeed_records, recordSize);

        //在读的时候进行统计waitCounter即可，因为写（pull）的时候可能正在阻塞，但读的时候已经能读到这个阻塞的counter数

        state.setLong(Messages.wait_reader_time, waitReaderTime);
        state.setLong(Messages.wait_writer_time, waitWriterTime);

        boolean isChannelByteSpeedLimit = (this.byteSpeed > 0);
        boolean isChannelRecordSpeedLimit = (this.recordSpeed > 0);
        if (!isChannelByteSpeedLimit && !isChannelRecordSpeedLimit)
        {
            return;
        }

        long lastTimestamp = lastState.getTimestamp();
        long nowTimestamp = System.currentTimeMillis();
        long interval = nowTimestamp - lastTimestamp;
        if (interval - this.flowControlInterval >= 0)
        {
            long byteLimitSleepTime = 0;
            long recordLimitSleepTime = 0;
            if (isChannelByteSpeedLimit)
            {


            }

            if (isChannelRecordSpeedLimit)
            {
                long currentRecordSpeed = (Messages.getTotalReadRecords(state) -
                        Messages.getTotalReadRecords(lastState)) * 1000 / interval;
                if (currentRecordSpeed > this.recordSpeed)
                {
                    // 计算根据recordLimit得到的休眠时间
                    recordLimitSleepTime = currentRecordSpeed * interval / this.recordSpeed
                            - interval;
                }
            }

            // 休眠时间取较大值
            long sleepTime = byteLimitSleepTime < recordLimitSleepTime ?
                    recordLimitSleepTime : byteLimitSleepTime;
            if (sleepTime > 0)
            {
                try
                {
                    Thread.sleep(sleepTime);
                } catch (InterruptedException e)
                {
                    Thread.currentThread().interrupt();
                }
            }


            lastState.setLong(Messages.read_succeed_records,
                    state.getLong(Messages.read_succeed_records));
            lastState.setLong(Messages.read_failed_records,
                    state.getLong(Messages.read_failed_records));
            lastState.setTimestamp(nowTimestamp);
        }
    }

    private void statPull(long recordSize)
    {
        state.increase(Messages.write_received_records, recordSize);

    }

    public void close()
    {
        this.isClosed = true;

        try
        {
            this.recordQueue.put(TerminateRecord.get());
        } catch (InterruptedException ex)
        {
            Thread.currentThread().interrupt();
        }
    }


    public void clear()
    {
        this.recordQueue.clear();
    }


    protected void put(Record record)
    {
        try
        {
            long startTime = System.nanoTime();
            this.recordQueue.put(record);
            waitWriterTime += System.nanoTime() - startTime;

        } catch (InterruptedException ex)
        {
            Thread.currentThread().interrupt();
        }
    }


    protected void addAll(Collection<Record> recordCollection)
    {
        try
        {
            long startTime = System.nanoTime();
            lock.lockInterruptibly();


            this.recordQueue.addAll(recordCollection);
            waitWriterTime += System.nanoTime() - startTime;

            notEmpty.signalAll();
        } catch (InterruptedException e)
        {
            throw new RuntimeException(e);
        } finally
        {
            lock.unlock();
        }
    }


    protected Record take()
    {
        try
        {
            long startTime = System.nanoTime();
            Record record = this.recordQueue.take();
            waitReaderTime += System.nanoTime() - startTime;

            return record;
        } catch (InterruptedException e)
        {
            Thread.currentThread().interrupt();
            throw new IllegalStateException(e);
        }
    }


    protected void drainTo(Collection<Record> recordCollection)
    {
        assert recordCollection != null;
        recordCollection.clear();
        try
        {
            long startTime = System.nanoTime();
            lock.lockInterruptibly();
            while (this.recordQueue.drainTo(recordCollection, bufferSize) <= 0)
            {
                notEmpty.await(200L, TimeUnit.MILLISECONDS);
            }
            waitReaderTime += System.nanoTime() - startTime;


            notInsufficient.signalAll();
        } catch (InterruptedException e)
        {
            throw new RuntimeException(e);
        } finally
        {
            lock.unlock();
        }
    }


    public int size()
    {
        return this.recordQueue.size();
    }


}
